package com.guchenbo.flink.kafka;

import com.guchenbo.flink.Transaction;
import com.guchenbo.flink.TransactionCodec;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author guchenbo
 * @date 2024/5/31
 */
public class FlinkKafkaDemo {
    private static final Logger log = LoggerFactory.getLogger(FlinkKafkaDemo.class);

    public static void main(String[] args) throws Exception {
        String inputTopic = "gcb-datagen";
        String outputTopic = "flink_output";
        String consumerGroup = "gcb-flink";
        String bootstrapServers = "kafka:9092";
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        // source
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers(bootstrapServers)
                        .setTopics(inputTopic).setGroupId(consumerGroup).setStartingOffsets(OffsetsInitializer.earliest())
                        .setValueOnlyDeserializer(new SimpleStringSchema()).build();
        DataStream<String> dataStream = environment.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(),
                        "Kafka Source");
        // 算子
        TransactionCodec transactionCodec = new TransactionCodec();
        DataStream<Transaction> transactionDataStream = dataStream.map(transactionCodec::fromCsv);
//        Transaction transaction = new Transaction();
//        transaction.getAccountId();
//        dataStream.keyBy(tran -> tran.getAccountId());

        // sink PrintSinkFunction
        transactionDataStream.print();
        log.info("starting kafka source");
        environment.execute("flink-kafka");
    }
}
